package com.webull.information.center.carwler.job;

import java.net.InetAddress;
import java.util.Collections;
import java.util.List;
import java.util.Optional;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.webull.information.center.carwler.common.model.CollectRecord;
import com.webull.information.center.carwler.common.model.parm.CollectParm;
import com.webull.information.center.carwler.common.util.eljob.ElasticJobUtil;
import com.webull.information.center.carwler.dao.CollectRecordDao;
import com.webull.information.center.carwler.task.News_GoogleFull_EtfFund_Task;
import com.webull.securities.client.model.Region;
import com.webull.securities.client.service.RegionService;

/**
 * 个股新闻(google)采集数据重建,按照交易所分片
 *
 * @author shimingjun
 * @version 1.0
 * @since JDK 1.8
 */
@Component
public class News_GoogleFull_EtfFund_Job implements SimpleJob {

    protected final Logger logger = LogManager.getLogger(getClass());

	private @Autowired News_GoogleFull_EtfFund_Task full_EtfFund_Task;

	private @Autowired RegionService regionService;

	private @Autowired CollectRecordDao collectRecordDao;

    private final String collectJob = "EtfFundNewsFullShardExchange";

	@Override
	public void execute(ShardingContext context) {
		int sharding = Optional.ofNullable(context).map(v -> v.getShardingItem()).orElse(-1);
		int totalCount = Optional.ofNullable(context).map(v -> v.getShardingTotalCount()).orElse(-1);

		logger.info("News_GoogleFull_EtfFund_Job start:" + ArrayUtils.toString(sharding) + ",totalCount:" + totalCount);
		List<Region> regions = regionService.searchAll();
		logger.info("News_GoogleFull_EtfFund_Job div size{};regions:{};", regions != null ? regions.size() : 0,
				ArrayUtils.toString(regions));
		List<Region> shardRegions = Optional.ofNullable(ElasticJobUtil.shardingItems(context, regions)).orElse(regions);
		process_0(shardRegions);
	}


    public void process_0(List<Region> shardRegions) {
        Collections.shuffle(shardRegions);
        logger.info("News_GoogleFull_EtfFund_Job div size{};shardRegion:{};", shardRegions != null ? shardRegions.size() : 0, ArrayUtils.toString(shardRegions));
        for (int i = 0; shardRegions != null && i < shardRegions.size(); i++) {
            try {
                List<CollectRecord> records = collectRecordDao.search(collectParm(shardRegions.get(i)));
                Integer lastId = getLastTickeId(records);
                logger.info("News_GoogleFull_EtfFund_Job shardRegion start collectJob{} div ,exchange:{},lastId{}", collectJob, shardRegions.get(i).getAlias(), lastId);
                Integer lastId_ = full_EtfFund_Task.fullEtfFundGoogleNewsRegion(shardRegions.get(i).getId(), -1);
                CollectRecord record = backRecord(shardRegions.get(i));
                record.setLastTickerId(lastId_);
                Integer collId = collectRecordDao.add(record);
                logger.info("News_GoogleFull_EtfFund_Job shardRegion end collectJob{} div ,exchange:{},collId{}", collectJob, shardRegions.get(i).getAlias(), collId);
            } catch (Exception e) {
                logger.error("News_GoogleFull_EtfFund_Job shardRegion collectJob div error:" + shardRegions.get(i), e);
            }
        }
    }

    private Integer getLastTickeId(List<CollectRecord> records) {
        if (records == null || records.isEmpty()) {
            return -1;
        } else {
            return records.get(0).getLastTickerId();
        }
    }

    private CollectParm collectParm(Region region) {
        CollectParm parm = new CollectParm();
        parm.setCollectJob(collectJob);
        parm.setExchangeCode(region.getAlias());
        parm.setExchangeId(region.getId());

        parm.setRecordType(CollectRecord.RecordTypeConstants.NEWS);
        return parm;
    }

    private CollectRecord backRecord(Region region) {
        CollectRecord record = new CollectRecord();
        record.setCollectJob(collectJob);
        record.setExchangeCode(region.getAlias());
        record.setExchangeId(region.getId());
        record.setRecordType(CollectRecord.RecordTypeConstants.NEWS);
        try {
            record.setCustomer(InetAddress.getLocalHost().getHostAddress());
        } catch (Exception e) {
        }
        return record;
    }
}
